package mapreduce

import (
	"encoding/json"
	"io/ioutil"
	"os"
)

func DoMap(
	jobName string,			//name of current job
	mapTaskNumber int,		//which reduce task this is
	inFile string,
	nReduce int,			//the number of reduce that were run("M" in paper)
	mapF func(file string, contents string) []KeyValue,		//map function from user code
){
	//read string data from input inFile
	byteData, _ := ioutil.ReadFile(inFile)
	contents := string(byteData)

	//invoke mapF function with input filename  and contents as arguments
	//assign return slice to a new slice
	retPairs := mapF(inFile, contents)

	//write slice to intermediate inFile
	interFileList := make([]*os.File, nReduce)
	encList := make([]*json.Encoder, nReduce)
	for i:= 0; i<nReduce; i++ {
		//partition inFile by reduceName function
		interFilename := reduceName(jobName, mapTaskNumber, i)
		file, _ := os.Create(interFilename)
		interFileList[i] = file
		encList[i] = json.NewEncoder(file)
		//fmt.Printf("mapTask#%d produce intermediate inFile: %s\n", mapTaskNumber, interFilename)
	}

	for _, pair := range retPairs {
		reduceTaskNo := ihash(pair.Key) % nReduce
		encList[reduceTaskNo].Encode(pair)
	}

	for _, file := range interFileList {
		file.Close()
	}
}
